package com.hudsonmq.spring.client;


import com.hudson.codec.MessageCodecFactory;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudsonmq.spring.config.HudsonMQConfig;
import com.hudsonmq.spring.handler.netty.BrokerInfoHandler;
import com.hudsonmq.spring.utils.NettyUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @Author: 渔小鲤
 * @DateTime: 2025/7/12 20:18
 **/
@Component
public class NameSrvClient implements InitializingBean {
    private static final Logger logger = LoggerFactory.getLogger(NameSrvClient.class);

    @Resource
    private HudsonMQConfig hudsonMQConfig;

    private static Bootstrap client;


    public static void sendBrokerAddress( Map<String, Channel> map) {
        for (Map.Entry<String, Channel> channelEntry : map.entrySet()) {
            NettyUtil.write(channelEntry.getValue(), RemotingCommand.builder()
                    .type(MessageTypeEnum.GET_BROKER_INFO.getType())
                    .build());
        }
    }

    public static void start(List<InetSocketAddress> nameSrvAddressList, Map<String, Channel> map) {
        for (InetSocketAddress inetSocketAddress : nameSrvAddressList) {
            try {
                map.put(inetSocketAddress.getAddress().getHostAddress() + inetSocketAddress.getPort(),
                        client.connect(inetSocketAddress).sync().channel());
                logger.info("连接 NameSrv {}成功", inetSocketAddress.getAddress().getHostAddress() + ":" + inetSocketAddress.getPort());
            } catch (InterruptedException e) {
                logger.error("连接NameSrv: {} 异常", inetSocketAddress.getAddress().getHostAddress() + ":" + inetSocketAddress.getPort());
                throw new RuntimeException(e);
            }
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        client = new Bootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel sc) throws Exception {
                        sc.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024 * 1024, 8, 4, 0, 0));
                        sc.pipeline().addLast(MessageCodecFactory.getCodec(hudsonMQConfig.getSerializableMethod()));
                        sc.pipeline().addLast(new IdleStateHandler(0, 10, 0, TimeUnit.SECONDS)); // 写空闲4秒触发 WRITE_IDLE
                        sc.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
                                if (evt instanceof IdleStateEvent) {
                                    IdleStateEvent event = (IdleStateEvent) evt;
                                    if (event.state() == IdleState.WRITER_IDLE) {
                                        // 发送心跳包
                                        logger.debug("produce 空闲，发送心跳至 broker...");
                                        ctx.writeAndFlush(RemotingCommand.builder().type(MessageTypeEnum.IDLE_MESSAGE.getType()).build()); // 或者发送自定义心跳消息
//                                                new Message<>(MessageTypeEnum.IDLE_MESSAGE.getType(), null)); // 或者发送自定义心跳消息
                                    }
                                }
                            }
                        });
                        sc.pipeline().addLast(new BrokerInfoHandler());
                    }
                });
    }
}
